package com.dc.rxjava;

import android.util.Log;

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiConsumer;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Function;

/**
 * 组合操作符
 *
 * @author 冰柠檬dc
 */
public class ComposeOperatorUtils {

    private static final String TAG = "RxJava";

    /**
     * concat操作符：可以将多个观察者组合在一起，然后按照之前发送顺序发送事件。需要注意的是，concat() 最多只可以发送4个事件。
     * concatArray操作符：与 concat() 作用一样，不过 concatArray() 可以发送多于 4 个被观察者。
     *
     * @return Disposable
     */
    public static Disposable concat() {
        return Observable.concat(Observable.just(1, 2).delay(2, TimeUnit.SECONDS), Observable.just(5, 6).delay(1, TimeUnit.SECONDS), Observable.just(8, 9)).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:5
        // onNext:6
        // onNext:8
        // onNext:9
        // onComplete
    }

    /**
     * merge操作符：这个方法与concat() 作用基本一样，只是 concat() 是串行发送事件，而 merge() 并行发送事件。
     * mergeArray操作符：与 merge() 的作用是一样的，只是它可以发送4个以上的被观察者
     *
     * @return Disposable
     */
    public static Disposable merge() {
        return Observable.merge(Observable.just(1, 2).delay(2, TimeUnit.SECONDS), Observable.just(5, 6).delay(1, TimeUnit.SECONDS), Observable.just(8, 9)).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:8
        // onNext:9
        // onNext:5
        // onNext:6
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * concatArrayDelayError操作符：在 concatArray() 和 mergeArray() 两个方法当中，如果其中有一个被观察者发送了一个 Error 事件，那么就会停止发送事件，
     * 如果你想 onError() 事件延迟到所有被观察者都发送完事件后再执行的话，就可以使用  concatArrayDelayError() 和 mergeArrayDelayError()
     *
     * @return Disposable
     */
    public static Disposable concatArrayDelayError() {
        return Observable.concatArrayDelayError(Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onError(new NumberFormatException());
            }
        }), Observable.just(5, 6)).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:5
        // onNext:6
        // onError
    }

    /**
     * zip操作符：会将多个被观察者合并，根据各个被观察者发送事件的顺序一个个结合起来，最终发送的事件数量会与源 Observable 中最少事件的数量一样。
     *
     * @return Disposable
     */
    public static Disposable zip() {
        return Observable.zip(Observable.intervalRange(1, 3, 2, 1, TimeUnit.SECONDS).map(new Function<Long, String>() {
            @Override
            public String apply(Long aLong) throws Exception {
                String s1 = "A" + aLong;
                Log.e(TAG, "A 发送的事件 " + s1);
                return s1;
            }
        }), Observable.intervalRange(1, 2, 0, 1, TimeUnit.SECONDS).map(new Function<Long, String>() {
            @Override
            public String apply(Long aLong) throws Exception {
                String s2 = "B" + aLong;
                Log.e(TAG, "B 发送的事件 " + s2);
                return s2;
            }
        }), new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Exception {
                return s + s2;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // B 发送的事件 B1
        // B 发送的事件 B2
        // A 发送的事件 A1
        // onNext:A1B1
        // A 发送的事件 A2
        // onNext:A2B2
        // onComplete
    }

    /**
     * combineLatest操作符：combineLatest() 的作用与 zip() 类似，但是 combineLatest() 发送事件的序列是与发送的时间线有关的，
     * 当 combineLatest() 中所有的 Observable 都发送了事件，只要其中有一个 Observable 发送事件，这个事件就会和其他 Observable 最近发送的事件结合起来发送
     * combineLatestDelayError() 就是多了延迟发送 onError() 功能
     *
     * @return Disposable
     */
    public static Disposable combineLatest() {
        return Observable.combineLatest(Observable.intervalRange(1, 3, 3, 2, TimeUnit.SECONDS).map(new Function<Long, String>() {
            @Override
            public String apply(Long aLong) throws Exception {
                String s1 = "A" + aLong;
                Log.e(TAG, "A 发送的事件 " + s1);
                return s1;
            }
        }), Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS).map(new Function<Long, String>() {
            @Override
            public String apply(Long aLong) throws Exception {
                String s2 = "B" + aLong;
                Log.e(TAG, "B 发送的事件 " + s2);
                return s2;
            }
        }), new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Exception {
                return s + s2;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // B 发送的事件 B1
        // B 发送的事件 B2
        // B 发送的事件 B3
        // B 发送的事件 B4
        // A 发送的事件 A1
        // onNext:A1B4
        // B 发送的事件 B5
        // A 发送的事件 A2
        // onNext:A2B5
        // A 发送的事件 A3
        // onNext:A3B5
        // onComplete
    }

    /**
     * reduce操作符：将发送数据以一定逻辑聚合起来，这两个的区别在于 scan() 每处理一次数据就会将事件发送给观察者，而 reduce() 会将所有数据聚合在一起才会发送事件给观察者。
     *
     * @return Disposable
     */
    public static Disposable reduce() {
        return Observable.just(1, 3, 5).reduce(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer sum, Integer integer) throws Exception {
                Log.e(TAG, "sum:" + sum);
                Log.e(TAG, "integer:" + integer);
                return sum + integer;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        });
        // 打印结果：(前2个数据聚合之后，然后再与后1个数据进行聚合，一直到没有数据为止)
        // sum:1
        // integer:3
        // sum:4
        // integer:5
        // onNext:9
    }

    /**
     * collect操作符：将数据收集到数据结构当中。
     *
     * @return Disposable
     */
    public static Disposable collect() {
        return Observable.just(1, 3, 5).collect(new Callable<ArrayList<Integer>>() {
            @Override
            public ArrayList<Integer> call() throws Exception {
                return new ArrayList<>();
            }
        }, new BiConsumer<ArrayList<Integer>, Integer>() {
            @Override
            public void accept(ArrayList<Integer> integers, Integer integer) throws Exception {
                integers.add(integer);
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        });
        // 打印结果：
        // onNext:[1, 3, 5]
    }

    /**
     * startWith操作符：在发送事件之前追加事件，startWith() 追加一个事件，startWithArray() 可以追加多个事件。追加的事件会先发出。
     *
     * @return Disposable
     */
    public static Disposable startWith() {
        return Observable.just(5, 6).startWithArray(2,3).startWith(1).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:3
        // onNext:5
        // onNext:6
        // onComplete
    }

    /**
     * count操作符：返回被观察者发送事件的数量。
     *
     * @return Disposable
     */
    public static Disposable count() {
        return Observable.just(1,2,3).count().subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        });
        // 打印结果：
        // onNext:3
    }
}
